
                /*********************************************
                **                                          **
                **                                          **
                **               reduce scatter-r           **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/


/*
** execute with 6 processes
*/

#include "arch.h"
#include "A.utilities"

//user program

Coll_id cid1, cid2, cid3, cid_bar, cid_red1[2], cid_red2[2];

//on most homogenous systems, explicit setting 
//won't be necessary (try commenting the
//setting lines, you should get the same values
//for the cids).
void init(int, char **){
   cid1.set();
   cid2.set();
   cid3.set();
   cid_bar.set();
   cid_red1[0].set();
   cid_red1[1].set();
   cid_red2[0].set();
   cid_red2[1].set();
   fprintf(Sdf::trace, "node%d: cid1=%d, cid2=%d, cid3=%d, cid_bar=%d, cid_red1=%d, %d,cid_red2=%d, %d\n",
           Sdf::PROC_ID, 
           cid1.get(), cid2.get(), cid3.get(),cid_bar.get(), 
           cid_red1[0].get(), cid_red1[1].get(),
           cid_red2[0].get(), cid_red2[1].get());
   fflush(Sdf::trace);
}

ChanL *chan;

CommR *commr = 0;

class Proc: public S_Thread{

   public:
        int pid;
        int step;

        Proc(int p, int s)
                : S_Thread((Smf0_ptr)&Proc::body){pid = p; step = s;}

        void body(){
               fprintf(Sdf::trace, "node %d, in proc %d, at start, step %d\n",
                       Sdf::PROC_ID, pid, step);
               fflush(Sdf::trace);
               int i;
               for(i = 0; i < 4; i++){
                   reschedule();
               }

               if(step<1){

                   int i;

                   int from[27];
                   int to[27];
                   for(i=0; i<27; i++){
                      from[i] = i;
                      to[i] = 777;
                   }

                   int scatter_table[6];
                   if(Sdf::PROC_ID < 3){
                         scatter_table[0] = 2;
                         scatter_table[1] = 3;
                         scatter_table[2] = 4;
                         scatter_table[3] = 5;
                         scatter_table[4] = 6;
                         scatter_table[5] = 7;
                   }else{
                         scatter_table[0] = 5;
                         scatter_table[1] = 6;
                         scatter_table[2] = 7;
                         scatter_table[3] = 2;
                         scatter_table[4] = 3;
                         scatter_table[5] = 4;
                   }
                     //rank in scatter_table corresponds to rank into group.
                     //According to scatter table 5+6+7 items are reduced in
                     //the group <0, 1, 2> and scattered in <3, 4, 5> so that
                     //the processes respectively get: 5, 6 and 7 items.
                     //Conversely, 9 items are reduced in <3, 4, 5> and
                     //scattered in <0, 1, 2> so that the processes 
                     //respectively get: 2, 3, and 4 items.

                   if(pid == 1){

                      chan->recv();
                        //local synchronization between proc1 and proc2
                        //for commr built in proc2 to be available when
                        //proc1 starts processing

                      fprintf(Sdf::trace, "node %d, in proc %d ***before reduce at step %d\n",
                              Sdf::PROC_ID, pid, step);
                      fflush(Sdf::trace);

                        //The in comm1 and comm2 are separatly reduced.
                        //Scattered result appears in processes of 
                        //comm2 and comm1.
                      _Sdf<int>::reduce(from[0], to[0], scatter_table, 
                                        Opa<int>::SUM, cid_red1, commr);

                      chan->send();
                        //local synchronization between proc1 and proc2
                        //for commr built in proc2 to be available until
                        //proc1 finishes processing
                   }else{

                      Group *group1;
                      int rk1[3];
                      for(i = 0; i < 3; i++)
                         rk1[i] = i;
                      group1 = Sdf::incl(Sdf::COMML_WORLD, 3, rk1);
                      CommL *comm1 = group1->commit(cid1, Sdf::COMML_WORLD);

                      Group *group2;
                      int rk2[3];
                      for(i = 3; i < Sdf::PROC_NB; i++)
                         rk2[i-3] = i;
                      group2 = Sdf::incl(Sdf::COMML_WORLD, Sdf::PROC_NB-3, rk2);
                      CommL *comm2 = group2->commit(cid2, Sdf::COMML_WORLD);

                      Sdf::barrier(cid_bar);

                      commr = Sdf::newCommR(cid3, comm1, comm2, Sdf::COMML_WORLD);

                      chan->send();
                        //local synchronization between proc1 and proc2
                        //to keep proc1 waiting until building commr has
                        //completed.

                      fprintf(Sdf::trace, "node %d, in proc %d ***before reduce at step %d\n",
                              Sdf::PROC_ID, pid, step);
                      fflush(Sdf::trace);

                        //The in comm1 and comm2 are separatly reduced.
                        //Scattered result appears in processes of 
                        //comm2 and comm1.
                      _Sdf<int>::reduce(from[0], to[0], scatter_table, 
                                        Opa<int>::SUM, cid_red2, commr);

                      chan->recv();
                        //local synchronization between proc1 and proc2
                        //to prevent commr, comm1 and comm2 from being
                        //destroyed before proc1 finishes processing
                      delete comm1;
                      delete comm2;
                      delete commr;
                   }

                   fprintf(Sdf::trace, "node %d, in proc %d ***after reduce, to=%d,%d,%d,%d,%d,%d,%d,%d\n",
                           Sdf::PROC_ID, pid,
                           to[0], to[1], to[2], to[3], 
                           to[4], to[5], to[6], to[7]);
                   fflush(Sdf::trace);

                   Proc *proc1 = new Proc(pid*10 + 1, step+1);
                   Proc *proc2 = new Proc(pid*10 + 2, step+1);
                   S_Thread *proc_tabl[2] = {proc1, proc2};
                   int pri[2] = {1, 1};
                   par(2, proc_tabl, pri);
                   delete proc1;
                   delete proc2;
               }
               fprintf(Sdf::trace, "on node %d, proc %d the end at step %d\n",
                        Sdf::PROC_ID, pid, step);
               fflush(Sdf::trace);
        }
};

class Root_proc: public Thread{

   public:

        Root_proc()
                : Thread((Tmf0_ptr)&Root_proc::body){}

        void body(){

                fprintf(Sdf::trace, "node %d, in Root_proc::body1\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
                Proc *proc1 = new Proc(1, 0);
                Proc *proc2 = new Proc(2, 0);

                chan = new ChanL;
                Sdf::barrier(cid_bar);

                fprintf(Sdf::trace, "node %d, in Root_proc::body2\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
                S_Thread *proc_tabl[2] = {proc1, proc2};
                int pri[2] = {1, 1};
                par(2, proc_tabl, pri);
                fprintf(Sdf::trace, "node %d, in Root_proc::body4\n", Sdf::PROC_ID);
                fflush(Sdf::trace);

                Sdf::barrier(cid_bar);
                delete chan;
                delete proc1;
                delete proc2;
        }
};
//end of user program

MAIN(Root_proc, init, Sdf::empty, Sdf::empty)

